package FastLearningSpark.SparkSQL;//import org.apache.spark.SparkConf;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
import com.alibaba.fastjson.JSONObject;

/**
 * <pre>
 * Created with IntelliJ IDEA.
 * User: zhengzhi
 * Date: 2017/1/23
 * Time: 15:12
 * To change this template use File | Settings | File Templates.
 * </pre>
 *
 * @author Administrator
 */
public class SparkSQL {

    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("SparkSQLTest").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);

        JavaRDD<String> lines = sc.textFile("E:/testData/datalink.txt");

        /**
         * 将文本文件中的每一行转换成JSON字符串格式的
         */
        lines.map(new Function<String, String>() {
            @Override
            public String call(String line) throws Exception {
                String[] splits = line.split("\\t", -1);
                JSONObject json = new JSONObject();
                json.put("imei",splits[0]);
                json.put("idfa",splits[1]);
                json.put("mac",splits[2]);
                json.put("createAt",splits[3]);
                json.put("latitude",splits[4]);
                json.put("longitude",splits[5]);
                json.put("ip",splits[6]);
                json.put("ssid",splits[7]);

                return json.toJSONString();
            }
        });


        SQLContext sqlContext = new SQLContext(sc);

        //直接从json文件中读取数据
        DataFrame input = sqlContext.jsonFile("D:/testData/testweet.json");

        //Displays the content of the DataFrame to stdout
        input.show();

        //Print the schema in a tree format
        input.printSchema();

        //注册输入的DataFrame，把读入的数据注册为一张临时表并赋予该表一个名字
        input.registerTempTable("tweets");

        //依据retweetCount（转发计数）选出推文
        DataFrame topTweets = sqlContext.sql("SELECT text,retweetCount FROM tweets ORDER BY retweetCount LIMIT 10");

        JavaRDD<String> topTweetText = topTweets.toJavaRDD().map(new Function<Row, String>() {
            @Override
            public String call(Row row) throws Exception {
                return row.getString(0);
            }
        });

        topTweetText.foreach(new VoidFunction<String>() {
            @Override
            public void call(String s) throws Exception {
                System.out.println(s);
            }
        });



        topTweets.show();


    }
}
